package com.li.flink.datastream.watermark

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala.{OutputTag, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object EventTimeWatermarkApp {
  def main(args: Array[String]): Unit = {
    //获取运行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //test01(env)
    test02(env)
    env.execute("WindowApp")
  }

  //基于EventTime和Watermark结合滚动窗口综合案例之没有延迟
  def test02(env: StreamExecutionEnvironment): Unit = {
    //连接socket
    var text = env.socketTextStream("127.0.0.1", 9001)
    val line = text.assignTimestampsAndWatermarks(
      new BoundedOutOfOrdernessTimestampExtractor[String](Time.seconds(2)) {
        override def extractTimestamp(t: String) = t.split(",")(0).toLong
      })
    val stream = line.map(m => {
      val splits = m.split(",")
      (splits(1), splits(2).toInt)
    })
    val reduceFunction = new ReduceFunction[Tuple2[String, Int]] {
      override def reduce(t: (String, Int), t1: (String, Int)): (String, Int) = {
        println("reduce invoke" + t._1 + ":" + (t1._2 + t._2))
        (t._1, t1._2 + t._2)
      }
    }

    class MyProcessWindowFunction extends ProcessWindowFunction[(String, Int), String, String, TimeWindow] {
      def process(key: String, context: Context, input: Iterable[(String, Int)], out: Collector[String]) = {
        input.foreach(i => {
          out.collect("sum value :" + i._1 + ":" + i._2)
        })
      }
    }
    val outputData = new OutputTag[(String, Int)]("test")
    val window = stream.keyBy(_._1).window(TumblingEventTimeWindows.of(Time.seconds(5)))
    window.sideOutputLateData(outputData)
    window.reduce(reduceFunction, new MyProcessWindowFunction())
    stream.print()

    val outputTag = stream.getSideOutput(outputData)
    outputTag.printToErr()
  }

  //基于EventTime和Watermark结合滚动窗口综合案例之有延迟
  def test01(env: StreamExecutionEnvironment): Unit = {
    //连接socket
    var text = env.socketTextStream("127.0.0.1", 9001)
    val line = text.assignTimestampsAndWatermarks(
      new BoundedOutOfOrdernessTimestampExtractor[String](Time.seconds(0)) {
        override def extractTimestamp(t: String) = t.split(",")(0).toLong
      })
    val stream = line.map(m => {
      val splits = m.split(",")
      (splits(1), splits(2).toInt)
    })
    stream.keyBy(_._1).window(TumblingEventTimeWindows.of(Time.seconds(5))).sum(1).print()
  }

}
